package com.cssl;

import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
import com.alibaba.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.alibaba.ververica.cdc.debezium.DebeziumSourceFunction;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @Author: chen
 * @Date: 2021/11/10 21:26
 * @Desc: 使用 flinkcdc 事实获取mysql数据
 */
public class FlinkCDC {
    public static void main(String[] args) throws Exception {
        System.setProperty("HADOOP_USER_NAME", "atguigu");
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //开启状态后端
        env.setStateBackend(new FsStateBackend("hdfs://hadoop102:9820/gmall-flink/ck"));
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(10000);
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000);
        //1.10版本默认重启次数是 int 的最大值,1.12版本可以不用配置
        //env.setRestartStrategy(RestartStrategies.fixedDelayRestart(重启次数,重启间隔));

        //cdc source
        DebeziumSourceFunction<String> sourceFunction = MySQLSource.<String>builder()
                .hostname("hadoop102")
                .port(3306)
                .username("root")
                .password("root")
                .databaseList("gmall-flink")
                .tableList("gmall-flink.base_trademark")
                //.deserializer(new StringDebeziumDeserializationSchema())
                //使用自定义序列化器
                .deserializer(new CustomerDeserialization())
                /*
                 * initial: 先全量,然后增量(那么如果做全量时,数据修改怎么保证一致,2.0之前是用锁实现)
                 * earliest: 从binlog开始的位置读(需要先开启binlog,然后建库,不然会报错)
                 * latest: 从binlog最新的位置读
                 * 还有指定 offset timestamp
                 * */
                .startupOptions(StartupOptions.initial())
                .build();

        //打印
        env.addSource(sourceFunction)
                .print();

        env.execute("flink-cdc");
    }
}
